package edu.wisc.cloudservice;

import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
import com.googlecode.protobuf.pro.duplex.CleanShutdownHandler;
import com.googlecode.protobuf.pro.duplex.PeerInfo;
import com.googlecode.protobuf.pro.duplex.RpcClientChannel;
import com.googlecode.protobuf.pro.duplex.RpcConnectionEventNotifier;
import com.googlecode.protobuf.pro.duplex.client.DuplexTcpClientBootstrap;
import com.googlecode.protobuf.pro.duplex.execute.ThreadPoolCallExecutor;
import com.googlecode.protobuf.pro.duplex.listener.RpcConnectionEventListener;
import com.googlecode.protobuf.pro.duplex.logging.CategoryPerServiceLogger;

import edu.wisc.cloudservice.CloudManager.CloudService.BlockingInterface;

import edu.wisc.cloudservice.CloudManager.StartBrokerRequest;
import edu.wisc.cloudservice.CloudManager.StartBrokerResponse;

import edu.wisc.cloudservice.CloudManager.StartConsumerRequest;
import edu.wisc.cloudservice.CloudManager.StartConsumerResponse;


public class CloudServiceClient {

	private static Log log = LogFactory.getLog(CloudServiceClient.class);
	private static CloudServiceClient _cloudServiceClient;

	private BlockingInterface cloudServiceInterface;
	private RpcController controller;

	private CloudServiceClient() {
		initializeClient();
	}

	private void initializeClient() {

		String serverHostname = "127.0.0.1";
		int serverPort = 5555;
		String clientHostname = "127.0.0.1";
		int clientPort = 5566;

		PeerInfo client = new PeerInfo(clientHostname, clientPort);
		PeerInfo server = new PeerInfo(serverHostname, serverPort);

		CleanShutdownHandler shutdownHandler = new CleanShutdownHandler();
		try {
			DuplexTcpClientBootstrap bootstrap = new DuplexTcpClientBootstrap(
					client, new NioClientSocketChannelFactory(
							Executors.newCachedThreadPool(),
							Executors.newCachedThreadPool()),
					new ThreadPoolCallExecutor(3, 10));

			// bootstrap.setCompression(compress);

			// RPC payloads are uncompressed when logged - so reduce logging
			CategoryPerServiceLogger logger = new CategoryPerServiceLogger();
			logger.setLogRequestProto(false);
			logger.setLogResponseProto(false);
			bootstrap.setRpcLogger(logger);

			shutdownHandler.addResource(bootstrap);

			// Set up the event pipeline factory.
			// setup a RPC event listener - it just logs what happens
			RpcConnectionEventNotifier rpcEventNotifier = new RpcConnectionEventNotifier();
			RpcConnectionEventListener listener = new RpcConnectionEventListener() {

				@Override
				public void connectionReestablished(
						RpcClientChannel clientChannel) {
					log.info("connectionReestablished " + clientChannel);
				}

				@Override
				public void connectionOpened(RpcClientChannel clientChannel) {
					log.info("connectionOpened " + clientChannel);
				}

				@Override
				public void connectionLost(RpcClientChannel clientChannel) {
					log.info("connectionLost " + clientChannel);
				}

				@Override
				public void connectionChanged(RpcClientChannel clientChannel) {
					log.info("connectionChanged " + clientChannel);
				}
			};
			rpcEventNotifier.setEventListener(listener);
			bootstrap.registerConnectionEventListener(rpcEventNotifier);

			RpcClientChannel channel = bootstrap.peerWith(server);

			cloudServiceInterface = CloudManager.CloudService
					.newBlockingStub(channel);
			controller = channel.newRpcController();

		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public static synchronized CloudServiceClient getCloudServiceClient() {
		if (_cloudServiceClient == null) {
			_cloudServiceClient = new CloudServiceClient();
		}
		return _cloudServiceClient;
	}

	public void scaleBrokerInstances(String dataCenter, int numInstances) {
		StartBrokerRequest request = StartBrokerRequest.newBuilder()
				.setDataCenter(dataCenter).setNumInstances(numInstances)
				.build();

		try {
			StartBrokerResponse response = cloudServiceInterface.startBroker(
					controller, request);
			log.info(" Response from Server : " + response.getIsSuccess());
		} catch (ServiceException e) {
			log.warn("blocking call threw ServiceException. ", e);
		}
	}
	
	public void scaleConsumerInstances(String dataCenter, int numInstances) {
		StartConsumerRequest request = StartConsumerRequest.newBuilder()
				.setDataCenter(dataCenter).setNumInstances(numInstances)
				.build();

		try {
			StartConsumerResponse response = cloudServiceInterface
					.startConsumer(controller, request);
			log.info(" Response from Server : " + response.getIsSuccess());
		} catch (ServiceException e) {
			log.warn("blocking call threw ServiceException. ", e);
		}
	}
}
